/*	$Id: queue.c 1094 2010-05-12 00:53:46Z phrakt $	*/
/*
 * Copyright (c) 2008 Jean-Francois Brousseau <phrakt@gmail.com>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
 * THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL  DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#include <sys/types.h>
#include <sys/time.h>

#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>

#include "log.h"
#include "mem.h"
#include "util.h"
#include "thread.h"
#include "private.h"


/*
 * httpd_queue_alloc()
 *
 */
cnp_queue_t*
cnp_queue_alloc(unsigned int capacity)
{
	cnp_queue_t *queue;

	if ((queue = cnp_malloc(sizeof(*queue))) == NULL) {
		cnp_log_err(cnp_logchan, "failed to allocate queue");
		return (NULL);
	}

	queue->q_capacity = capacity;
	queue->q_count = 0;
	queue->q_head = queue->q_tail = 0;

	if ((queue->q_data = cnp_malloc(capacity * sizeof(void *))) == NULL) {
		cnp_log_err(cnp_logchan, "failed to allocate queue");
		cnp_free(queue);
		return (NULL);
	}

	if (cnp_mutex_init(&queue->q_lock, NULL) == -1) {
		cnp_log_err(cnp_logchan, "failed to initialize queue lock");
		cnp_free(queue->q_data);
		cnp_free(queue);
		return (NULL);
	}

	if (cnp_cond_init(&queue->q_cond, NULL) == -1) {
		cnp_log_err(cnp_logchan,
		    "failed to initialize queue condition");
		cnp_free(queue->q_data);
		cnp_free(queue);
		return (NULL);
	}

	return (queue);
}

/*
 * cnp_queue_free()
 *
 */
void
cnp_queue_free(cnp_queue_t *queue)
{
	(void)cnp_mutex_destroy(&queue->q_lock);
	(void)cnp_cond_destroy(&queue->q_cond);

	cnp_free(queue->q_data);
	cnp_free(queue);
}

/*
 * cnp_queue_put()
 *
 * Put the object <obj> at the end of the queue <queue>.
 * processing by one of the response threads.
 * Returns 0 on success, or -1 on failure.
 */
int
cnp_queue_put(cnp_queue_t *queue, void *obj)
{
	if (cnp_mutex_lock(&queue->q_lock) == -1) {
		cnp_log_err(cnp_logchan, "failed to lock queue");
		return (-1);
	}

	if (queue->q_count == queue->q_capacity) {
		cnp_log(cnp_logchan, CNP_LOG_WARN, "queue has reached maximum");
		(void)cnp_mutex_unlock(&queue->q_lock);
		return (-1);
	}

	queue->q_data[queue->q_tail] = obj;
	queue->q_tail = (queue->q_tail + 1) % queue->q_capacity;
	queue->q_count++;

	(void)cnp_cond_signal(&queue->q_cond);
	(void)cnp_mutex_unlock(&queue->q_lock);

	return (0);
}

/*
 * cnp_queue_get()
 *
 * Remove the next available object from the queue <queue>, with an optional
 * timeout specified in <timeout>, and return that object.  If <timeout> is
 * a null pointer, the call will wait indefinitely for a new object.
 *
 * Returns a pointer to the dequeued object on success, or NULL if an error
 * occurred.
 */
void*
cnp_queue_get(cnp_queue_t *queue, const struct timeval *timeout)
{
	int ret = 0;
	void *obj = NULL;
	struct timeval now;
	struct timespec ts;

	if (cnp_mutex_lock(&queue->q_lock) == -1) {
		cnp_log_err(cnp_logchan, "failed to lock queue");
		return (NULL);
	}

	if (queue->q_count == 0) {
		if (timeout != NULL) {
			(void)gettimeofday(&now, NULL);
			timeradd(&now, timeout, &now);
			TIMEVAL_TO_TIMESPEC(&now, &ts);

			ret = cnp_cond_timedwait(&queue->q_cond,
			    &queue->q_lock, &ts);
		}
		else {
			ret = cnp_cond_wait(&queue->q_cond, &queue->q_lock);
		}

		if ((ret == -1) && (cnp_error_get() != CNP_ERR_TIMEOUT))
			cnp_log_err(cnp_logchan, "failed to wait for next "
			    "available request");
	}

	if ((ret == 0) && (queue->q_count > 0)) {
		obj = queue->q_data[queue->q_head];
		queue->q_head = (queue->q_head + 1) % queue->q_capacity;
		queue->q_count--;
	}

	(void)cnp_mutex_unlock(&queue->q_lock);

	return (obj);
}

/*
 * cnp_queue_getsize()
 *
 * Returns the number of elements currently stored in the queue <queue>.
 */
unsigned int
cnp_queue_getsize(const cnp_queue_t *queue)
{
	return (queue->q_count);
}
